package com.wedata.RabbitMq;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class MQUtil {

    /**
     * 创建连接
     * @return
     * @throws IOException
     * @throws TimeoutException
     */
    public static Connection getConnection() throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();//MQ采用工厂模式来完成连接的创建
        //2.在工厂对象中设置连接信息(ip,port,virtualhost,username,password)
        /*factory.setHost("47.95.112.14");//设置MQ安装的服务器ip地址
        factory.setPort(5675);//设置端口号*/
        factory.setHost("trabbitmq.wegymer.cn");//设置MQ安装的服务器ip地址
        factory.setPort(5675);//设置端口号
        factory.setVirtualHost("wedata");//设置虚拟主机名称
        //MQ通过用户来管理
        factory.setUsername("wedata");//设置用户名称
        factory.setPassword("mtjZP42v4r8DCf3");//设置用户密码
        //3.通过工厂对象获取连接
        Connection connection = factory.newConnection();
        return connection;
    }


    /**
     * 生产者推送消息
     * @throws IOException
     * @throws TimeoutException
     */

    public static void producer(Integer num) throws IOException, TimeoutException {
        //获取连接
        Connection connection = MQUtil.getConnection();
        //mq提供Channel来将处理消息
        //创建Channel
        Channel channel = connection.createChannel();

        String msg = "hello world!>>你好 世界！"+num;
        //basicPublish将消息发送到指定的交换机
        //channel.basicPublish("ex3", "a", null, msg.getBytes());

        channel.basicPublish("wedata-test", "dev", null, msg.getBytes());
        //channel.basicPublish("wedata-test", "test", null, msg.getBytes());
        //channel.basicPublish("wedata-test", "pre", null, msg.getBytes());

        //关闭连接
        channel.close();
        //connection.close();

    }
    /**
     * 生产者推送消息(超时时间TTL)
     * @throws IOException
     * @throws TimeoutException
     */

    public static void producer(String time) throws IOException, TimeoutException {
        //获取连接
        Connection connection = MQUtil.getConnection();
        //mq提供Channel来将处理消息
        //创建Channel
        Channel channel = connection.createChannel();

        String msg = "hello world!>>你好 世界！"+time;
        //basicPublish将消息发送到指定的交换机
        //channel.basicPublish("ex3", "a", null, msg.getBytes());

        //channel.basicPublish("wedata-test", "dev", null, msg.getBytes());
        //channel.basicPublish("wedata-test", "test", null, msg.getBytes());
        //channel.basicPublish("wedata-test", "pre", null, msg.getBytes());

        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
        builder.expiration(time);
        channel.basicPublish("wedata-test", "dle", builder.build(), msg.getBytes());

        //关闭连接
        channel.close();
        //connection.close();

    }

    /**
     * 消息消费者从消息队列中获取消息
     */
    public static void consumer() throws IOException, TimeoutException {
        //获取与MQ的连接
        Connection connection = MQUtil.getConnection();
        //创建Channel
        Channel channel = connection.createChannel();
        //通过basicConsumer方法从指定队列中获取消息，消息生产者会通过ex2交换机中的key值将消息发送到queue6中，因为在创建queue6时绑定到交换机ex3中，指定的路由键为a。

        //获取Consumer
        Consumer consumer = new DefaultConsumer(channel){
            //重写handleDelivery方法（这个方法是消息的处理过程）
            //body参数就是接受到的消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //将消息转换成String类型然后打印
                String msg = new String(body);
                System.out.println(msg);
            }
        };
        //consumer参数是消息接收之后的处理方法
        channel.basicConsume("wedata-test",true,consumer);
    }

    public static void main(String[] args) throws IOException, TimeoutException {
        //producer();
        consumer();
    }
}
